package com.hung.triple.consumer;

import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.hung.triple.bootstrap.TripleConsumerConfig;
import com.hung.triple.bootstrap.config.TripleRegistryConfig;
import com.hung.triple.common.enums.TripleRegisterType;
import com.hung.triple.common.exception.TripleRpcException;
import com.hung.triple.consumer.response.FuturePool;
import com.hung.triple.consumer.response.TripleFuture;
import com.hung.triple.protocol.TripleProtocol;
import com.hung.triple.protocol.rpc.TripleRpcRequest;
import com.hung.triple.protocol.rpc.TripleRpcResponse;
import com.hung.triple.registry.ProviderAddressPool;
import com.hung.triple.registry.RegisterFactory;
import com.hung.triple.registry.nacos.NacosUtil;
import com.hung.triple.transport.Transporter;
import com.hung.triple.transport.TransporterPool;
import com.hung.triple.transport.netty4.client.NettyConnectionFactory;
import io.netty.channel.Channel;
import lombok.AllArgsConstructor;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 远程调用初始化 Bean
 *
 * @author Hung
 * @since 2022/9/18
 */
@Slf4j
public class TripleReferenceBean<T> {

    private transient volatile T ref;

    // 注册中心配置
    @Setter
    private TripleRegistryConfig registryConfig;
    @Setter
    private String groupName;
    @Setter
    private String version;
    @Setter
    private Class<?> interfaceClass;
    @Setter
    private String interfaceName;
    // 同步/异步
    @Setter
    private boolean async;


    public TripleReferenceBean() {
        groupName = "Triple";
        version = "1.0.0";
        async = false;
    }

    public TripleReferenceBean(String groupName, String version, Boolean async) {
        this.groupName = groupName;
        this.version = version;
        this.async = async;
    }

    @SuppressWarnings("unchecked")
    public T get() throws Exception {
        if (ref == null) {
            synchronized (this) {
                if (ref == null) {
                    // 参数校验
                    check();
                    // 初始化连接
                    initConnection();
                    // 生成代理对象
                    T t = (T) genProxyObject();
                    ref = t;
                }
            }
        }
        return ref;
    }

    private void initConnection() throws NacosException {

        // 1. 从注册中心获取需要连接的远程服务器地址，并保存进ProviderAddressPool.POOL中
        List<ConnectionAddressBeanInfo> needConnectionAddress = new ArrayList<>();
        switch (TripleRegistryConfig.getRegisterType()) {
            case ZOOKEEPER:
                // 注册中心模式下，只要有可用连接，就无需再次注册
                if (TransporterPool.hasAvailableTransporter()) {
                    return;
                }
                // TODO 连接 ZK & 读取 provider 信息 & 读取配置信息（序列化方式等）
                // 序列化方式两方默认演员  NacosUtil可以获取相应信息
                // 了解 HTTP client
                break;
            case DIRECT:
//                registryConfig.getAddressList().forEach(ads -> {
//                    Transporter transporter = TransporterPool.getTransporter(ads);
//                    if (transporter == null || !transporter.available()) {
//                        needConnectionAddress.add(ads);
//                    }
//                });
                break;
            case NACOS:
                //Client通过服务名找出再Nacos注册的服务，并将其缓存到needConnectionAddress中
                List<Instance> allInstance = NacosUtil.getAllInstance(interfaceClass.getSimpleName(), groupName, version);
                List<String> instancesList = allInstance.stream().map(instance -> {
                    String address = instance.getIp() + ":" + instance.getPort();
                    needConnectionAddress.add(new ConnectionAddressBeanInfo(address, interfaceClass.getSimpleName()));
                    return address;
                }).collect(Collectors.toList());
                ProviderAddressPool.add(groupName, interfaceClass.getSimpleName(), version, instancesList);
                break;
            default:
                break;
        }

        // 2. 建立与provider连接
        needConnectionAddress.forEach(connectionAddressBeanInfo -> {
            String[] split = connectionAddressBeanInfo.getRemoteAddress().split(":");
            String ip = split[0];
            int port = Integer.parseInt(split[1]);
            switch (TripleConsumerConfig.getProtocol()) {
                case TRIPLE_NETTY:
                    //没有Provider信息
                    //NettyConnectionFactory.rebuild();
                    try {
                        Channel channel = NettyConnectionFactory.getClient().connect(ip, port);
                        TransporterPool.addTransporter(connectionAddressBeanInfo.getServiceName(), connectionAddressBeanInfo.getRemoteAddress(), channel);
                    } catch (Exception e) {
                        log.info("获取连接错误");
                        throw new RuntimeException(e);
                    }
                    break;

                case TRIPLE_HTTP:
                    // TODO：HTTP
                default:
                    break;
            }
        });
    }

    /**
     * 生成代理对象
     */
    private Object genProxyObject() throws Exception {
        Class<?> ifClz;
        ClassLoader loader = Thread.currentThread().getContextClassLoader();
        if (interfaceClass != null) {
            ifClz = interfaceClass;
        } else if (!StringUtils.isEmpty(interfaceName)) {
            ifClz = Class.forName(interfaceName);
        } else {
            throw new IllegalArgumentException("triple can't found interfaceName");
        }

        /*
            创建代理对象，对返回的 Object 的任何操作，都会被转发到下述代码执行
              * proxy:代理类对象
              * method:被调用的被代理类方法
              * args：被调用的被代理类方法的入参
              * 返回值（Object）：调用的返回值
         */
        return Proxy.newProxyInstance(loader, new Class[]{ifClz},
                new ConsumerProxy(new ConsumerBeanInfo(groupName, version, ifClz.getSimpleName(), TripleConsumerConfig.getTimeoutMS(), async)));
    }

    private void check() {
        if (registryConfig == null) {
            throw new TripleRpcException("please set register info");
        }
    }

    /**
     * 生成 consumer 的代理对象
     *
     * @author Hung
     * @since 2022/9/18
     */
    @AllArgsConstructor
    class ConsumerProxy implements InvocationHandler {
        private final ConsumerBeanInfo consumerBean;

        @Override
        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

            String group = consumerBean.getGroupName();
            String serviceName = consumerBean.getServiceName();
            String version = consumerBean.getVersion();
            Long consumerTimeout = consumerBean.getTimeoutMS();
            TripleRegisterType registerType = TripleRegistryConfig.getRegisterType();

            //0.订阅服务集群变化
            RegisterFactory.getRegister().subscribe(consumerBean);

            // 1. 构造 RPC 请求
            TripleRpcRequest rpcRequest = new TripleRpcRequest();
            rpcRequest.setGroup(group);
            rpcRequest.setVersion(version);
            rpcRequest.setClassName(serviceName);
            rpcRequest.setMethodName(method.getName());
            rpcRequest.setParameterTypes(method.getParameterTypes());
            rpcRequest.setParameters(args);

            // 2. 构造异步响应结果（双写，pool + threadLocal）
            TripleFuture tripleFuture = FuturePool.push(rpcRequest);
            //设置过期时间
            long timeoutMicroSeconds = consumerTimeout == null ? TripleConsumerConfig.getTimeoutMS() : consumerTimeout;
            tripleFuture.setTimeoutMS(timeoutMicroSeconds);

            // 3. 发送 RPC 网络请求
            Transporter transporter = TransporterPool.getTransporter(registerType, serviceName);

            if (transporter == null) {
                //无连接 重试一次
                initConnection();
                transporter = TransporterPool.getTransporter(registerType, serviceName);
                if (transporter == null) {
                    throw new RuntimeException("该服务目前暂无连接");
                }
            }
            transporter.sendAsync(TripleProtocol.buildRpcRequest(rpcRequest));

            // 4. 处理返回结果
            if (consumerBean.isAsync()) {
                return null;
            }
            // 超时会抛出异常，继续向外抛
            TripleRpcResponse response = tripleFuture.get(timeoutMicroSeconds, TimeUnit.MILLISECONDS);
            switch (response.getCode()) {
                //成功提升权重，失败减低权重
                case TripleRpcResponse.SUCCESS:
                    transporter.promoteWeight();
                    return response.getResult();
                case TripleRpcResponse.INVOKE_SUCCESS_EXECUTE_FAILED:
                    transporter.lowerWeight();
                    throw response.getThrowable();
                default:
                    transporter.lowerWeight();
                    throw new TripleRpcException("invoke failed");
            }
        }
    }
}
